package com.yf.limiter.impl;

import com.yf.limiter.RateLimiter;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 漏桶算法
 * 核心：定时任务主动按固定速率处理请求，确保输出绝对平滑
 */
public class LeakyBucketRateLimiter<T> implements RateLimiter<T> {
    // 存储待处理请求的队列（阻塞队列，线程安全）
    private final BlockingQueue<T> requestQueue;
    // 定时出水的线程池（单线程，保证匀速）
    private final ScheduledExecutorService leakExecutor;
    // 漏桶是否已关闭
    private final AtomicBoolean isShutdown = new AtomicBoolean(false);

    /**
     * 构造函数
     * @param capacity 漏桶容量
     * @param ratePerSecond 出水速率（请求数/秒），支持小数（如2.5）
     */
    public LeakyBucketRateLimiter(int capacity, double ratePerSecond) {
        if (capacity <= 0) throw new IllegalArgumentException("容量必须大于0");
        if (ratePerSecond <= 0) throw new IllegalArgumentException("出水速率必须大于0");

        this.requestQueue = new LinkedBlockingQueue<>(capacity); // 有界队列，满了自动拒绝
        this.leakExecutor = Executors.newSingleThreadScheduledExecutor();


        long leakIntervalMs = (long) (1000.0 / ratePerSecond);
        leakExecutor.scheduleAtFixedRate(this::leak, 0, leakIntervalMs, TimeUnit.MILLISECONDS);
    }

    /**
     * 提交请求到漏桶（入桶）
     * @param request 待处理的请求（Runnable）
     */
    public boolean submit(T request) {
        if (isShutdown.get()) throw new IllegalStateException("漏桶已关闭");
        // 有界队列offer：队列满了直接返回false（拒绝请求）
        return requestQueue.offer(request);
    }

    /**
     * 主动出水：处理一个待请求（定时任务调用）
     */
    private void leak() {
        if (isShutdown.get()) return;
        // 从队列取出一个请求处理（队列为空时take会阻塞，这里用poll避免阻塞）
        T request = requestQueue.poll();
    }

}
